import rx.Observable;

/**
 * Created by czx on 2017/2/26.
 */
public class RunningAverage implements Observable.Transformer<Integer, Double> {

    private static class AverageAcc{
        int sum;
        int count;

        public AverageAcc(int sum, int count) {
            this.sum = sum;
            this.count = count;
        }
    }
    @Override
    public Observable<Double> call(Observable<Integer> integerObservable) {
        return integerObservable.scan(new AverageAcc(0,0),(acc,next)->new AverageAcc(acc.sum+next,acc.count+1))
                .filter(averageAcc1 -> averageAcc1.count>0)
                .map(averageAcc ->(averageAcc.sum/(double)averageAcc.count));
    }
}
